fs/aio: rework lio_listio() and fix AIO crashes and POSIX conformance - #20107

Previously, lio_listio() called aio_read()/aio_write() to submit the
I/O and only then initialized the per-request notification state
(aio_priv based), so a worker thread could complete an operation before
that state was set up (thread-unsafe), and the completion notification
hijacked the per-request sigevent machinery.

Rework the implementation: lio_listio() now links every aiocb of the
batch into a list (lio_link) before any I/O is submitted.  When an
operation completes, aio_signal() removes its node from the list under
aio_lock() and delivers the lio_listio completion notification only
when the list becomes empty.  The unused aio_priv field is replaced by
the lio_link/lio_sigevent/lio_sigwork fields in struct aiocb.

Co-developed-by: wushenhui <wushenhui@xiaomi.com>
Signed-off-by: wushenhui <wushenhui@xiaomi.com>
Signed-off-by: Xiang Xiao <xiaoxiang@xiaomi.com>

When lio_listio() is called with LIO_NOWAIT and a non-NULL sig, and no
I/O could be queued (or all entries are LIO_NOP/NULL), the completion
notification dereferences a NULL aiocbp picked from an empty iteration,
crashing nxsig_notification().

Scan the list for any non-NULL entry before delivering the
notification, and skip it entirely when the list contains only NULL
entries.

Signed-off-by: zhengyu16 <zhengyu16@xiaomi.com>

When a queued operation fails immediately (bad fd, EINVAL, or a failed
aio_read/aio_write submission), lio_listio() unconditionally deleted
the aiocbp from the request list.  In LIO_WAIT mode (or when no sig was
requested) the lio_link nodes were never linked into the list, so
list_delete() corrupted memory and crashed.

Only unlink the node when it was actually linked, i.e. when
mode == LIO_NOWAIT and a sigevent was provided.

Signed-off-by: tengshuangshuang <tengshuangshuang@xiaomi.com>

POSIX declares lio_listio() as:

  int lio_listio(int, struct aiocb *restrict const [restrict], int,
                 struct sigevent *restrict);

Update the prototype in include/aio.h (and the implementation and
libc.csv entry) accordingly, and drop the parameter names from the
other aio_* prototypes for consistency.

Signed-off-by: guoshichao <guoshichao@xiaomi.com>

lio_listio() never validated 'nent' against {AIO_LISTIO_MAX}, so a
batch larger than the documented limit was silently accepted, and the
hard-coded _POSIX_AIO_LISTIO_MAX value of 2 was too small for real
workloads (LTP uses 10 entries per call).

Add the FS_AIO_LISTIO_MAX Kconfig option (default 10), use it for
_POSIX_AIO_LISTIO_MAX in include/limits.h, validate 'nent' in
lio_listio(), and report the limit through sysconf(_SC_AIO_LISTIO_MAX).

Signed-off-by: tengshuangshuang <tengshuangshuang@xiaomi.com>

aioc_decant() frees the AIO container and detaches the aiocbp.  The
I/O workers (aio_read_worker, aio_write_worker, aio_fsync_worker)
called it before signaling completion, so aio_signal() and any code
touching the container afterwards ran on freed memory.  Additionally,
if the caller closed the file early the detached container could be
reused with a stale file reference.  Move aioc_decant() to after
aio_signal() and use aioc->aioc_aiocbp directly in the workers.

aio_cancel() also had two problems: with no aiocbp it looped over
g_aio_pending with a do/while that skipped the list re-entry check, so
a failed work_cancel() on an already running I/O caused an endless
loop; and an invalid fildes only checked 'fildes < 0' instead of
validating the descriptor, so a closed fd was not reported as EBADF.
Use a for-loop that always advances and validate the descriptor with
file_get()/file_put().

Co-developed-by: wushenhui <wushenhui@xiaomi.com>
Signed-off-by: wushenhui <wushenhui@xiaomi.com>
Signed-off-by: tengshuangshuang <tengshuangshuang@xiaomi.com>

Per POSIX, aio_read() and aio_write() must return -1 and set errno to
EINVAL when the request cannot be queued (aio_reqprio < 0,
aio_offset < 0), and the error must also be retrievable via
aio_error().  Conversely, when queuing fails with a bad file
descriptor, the error belongs to the asynchronous operation: the
functions must return 0 and report EBADF through aio_error().

- Merge the offset/reqprio checks and return ERROR with errno set,
  after storing the result in aio_result for aio_error().
- Drop the aio_fildes < 0 early return: a closed descriptor is now
  caught by fcntl()/aio_queue() and reported through aio_result with
  the function returning OK.
- aio_error(): report -EINVAL (failed validation) through errno
  instead of returning it as an error value.

Signed-off-by: tengshuangshuang <tengshuangshuang@xiaomi.com>

aio_suspend() checked the completion status once and then performed a
single sigtimedwait().  Any SIGPOLL delivered by an unrelated AIO
operation (one not referenced by 'list') woke the caller even though
none of the awaited requests had completed, and with a timeout the
remaining wait time was not preserved either.

Re-check the completion status after every wakeup and continue
waiting, recomputing the remaining time from the absolute deadline so
that the full timeout is honored.

Signed-off-by: wushenhui <wushenhui@xiaomi.com>

aio_fsync() never initialized aiocbp->lio_link, but the reworked
aio_signal() tests list_in_list(&lio_link) on every completion.  With
an uninitialized (or zero-filled) lio_link the behavior was
unpredictable; initialize the node so standalone fsync operations are
self-consistent.

Signed-off-by: tengshuangshuang <tengshuangshuang@xiaomi.com>

lio_listio() links each aiocbp->lio_link into its batch list before
submitting the I/O, but submitted the operations through the public
aio_read()/aio_write(), which re-initialized lio_link and destroyed
the list membership.  With an aiocb pre-filled with garbage (as in
ostest), the completion path then walked an invalid list.

Extract aio_read_internal()/aio_write_internal() that skip the
lio_link setup; aio_read()/aio_write() initialize lio_link (and
reject a NULL aiocbp) before calling the internal functions, while
lio_listio() calls the internal functions directly to preserve its
own lio_link setup.  For entries that are not part of a batch,
lio_listio() self-initializes lio_link instead.

Signed-off-by: Xiang Xiao <xiaoxiang@xiaomi.com>

aio_fsync()/aio_read()/aio_write()/lio_listio() initialized
aiocbp->lio_link with list_initialize(), which makes the node
self-referential (prev = next = &node).  aio_signal() tests
list_in_list(&lio_link) to detect lio_listio batches, so it wrongly
entered the lio_listio completion path for every standalone AIO
operation and notified through the uninitialized
lio_sigevent/lio_sigwork.

With CONFIG_SIG_EVTHREAD=y, garbage lio_sigevent.sigev_notify ==
SIGEV_THREAD caused nxsig_notification() to queue &lio_sigwork.work
onto the low-priority work queue with garbage func/value.  After the
aiocb was freed, the dangling work_s was dispatched with worker=NULL,
crashing in work_dispatch().

Fix: initialize lio_link with list_clear_node() (prev = next = NULL)
so list_in_list() returns false for non-lio_listio operations and
aio_signal() skips the lio_listio path.

While there, reject a NULL aiocbp in aio_fsync(): POSIX Issue 6 no
longer defines a NULL special case, and the old DEBUGASSERT() panicked
debug builds.

Co-developed-by: dengwenqi <dengwenqi@xiaomi.com>
Co-developed-by: fangxinyong <fangxinyong@xiaomi.com>
Signed-off-by: fangxinyong <fangxinyong@xiaomi.com>
Signed-off-by: dengwenqi <dengwenqi@xiaomi.com>
Signed-off-by: Xiang Xiao <xiaoxiang@xiaomi.com>